#include "EventLoopThreadPool.h"
#include "EventLoopThread.h"
#include "EventLoop.h"
#include "Logger.h"
#include "Timestamp.h"

#include <iostream>
#include <fstream>
#include <sstream>
#include <unistd.h>
#include <chrono>

EventLoopThreadPool::EventLoopThreadPool(EventLoop *base, int numThreads, const std::string& name)
  : m_baseLoop(base)
  , m_numThread(numThreads)
  , m_name(name)
  , m_next(0)
  , m_started(false)
{}
void EventLoopThreadPool::start(const ThreadInitCallback& cb)
{
  m_started = true;
  for (int i = 0; i < m_numThread; ++i) {
    char name[32];
    snprintf(name, sizeof(name), "%s%d", m_name.c_str(), i);
    m_threads.emplace_back(new EventLoopThread(cb, name));
    m_loops.push_back(m_threads.back()->startLoop()); 
  }
  if (0 == m_numThread && cb) {
    cb(m_baseLoop);
  }
}

EventLoop* EventLoopThreadPool::getNextLoop()
{
  EventLoop* loop = m_baseLoop;
  if (!m_loops.empty()) {
    loop = m_loops[m_next];
    ++m_next;
    if (m_next >= m_loops.size()) {
      m_next = 0;
    }
  }
  return loop;
}

EventLoop* EventLoopThreadPool::getNextLoop(const std::string& key)
{
  return getNextLoop();
} 

std::vector<EventLoop*> EventLoopThreadPool::getAllLoop()
{
  if (m_loops.empty()) {
    return {m_baseLoop};
  } else {
    return m_loops;
  }
}


DynamicEventLoopThreadPool::DynamicEventLoopThreadPool(EventLoop* base, int numThread, const std::string& name)
{
  m_baseLoop = base;
  m_name = name;
  m_started = false;
  m_monitoring.store(false);
  m_metrics.m_pendingTasks.store(0);
  m_metrics.m_cpuUsage.store(0.0);
  m_metrics.m_avgLatency.store(0);
  setThreadNum(numThread);
}

void DynamicEventLoopThreadPool::start(const ThreadInitCallback& cb)
{
  m_threadInitCallback = cb;
  for (int i = 0; i < m_currentThread.load(); i++) {
    m_workers.emplace_back(
      new EventLoopThread(cb, buildName(i)),
      true
    );
    m_loops.emplace_back(m_workers.back().m_thread->startLoop());
  }

  if (0 == m_currentThread && cb) {
    cb(m_baseLoop);
  }
}

EventLoop* DynamicEventLoopThreadPool::getNextLoop()
{
  // 基于加权随机算法的负载均衡
  std::lock_guard<std::mutex> lk(m_mutex);
  std::vector<std::pair<EventLoop*, int>> candidates;

  int totalWeight = 0;

  // 收集活跃线程及其权重
  for (auto& wt : m_workers) {
    if (wt.m_active.load()) {
      auto loop = wt.m_thread->getLoop();
      const int weight = 100 - std::min(100, (int)loop->getlistenCnt() / 10);
      candidates.emplace_back(loop, weight);
      totalWeight += weight;
    }
  }

  // 随机选择一个线程
  if (!candidates.empty()) {
    int rnd = rand() % totalWeight;
    for (auto& item : candidates) {
      if (rnd < item.second) {
        return item.first;
      }
      rnd -= item.second;
    }
  }

  // 降级到主循环
  return m_baseLoop;
}

EventLoop* DynamicEventLoopThreadPool::getNextLoop(const std::string& key)
{
  return this->getNextLoop();
}

std::string DynamicEventLoopThreadPool::buildName(const uint32_t index)
{
  char name[32];
  snprintf(name, sizeof(name), "%s%d", m_name.c_str(), index);
  return std::string(name);
}

void DynamicEventLoopThreadPool::setThreadNum(const int numThread)
{
  m_minThread.store(numThread);
  m_maxThread.store(2 * numThread);
  m_currentThread.store(numThread);
}

void DynamicEventLoopThreadPool::monitoringThread()
{
  while (m_monitoring.load()) {
    // 采集系统级指标
    const double cpu = getCpuUsage();
    const int64_t mem = getMemUsage();
    // 采集应用级指标
    int64_t totalPending = 0;
    {
      std::unique_lock<std::mutex> lock(m_mutex);
      for (auto& worker : m_workers) {
        if (worker.m_active) {
          totalPending += worker.m_thread->getLoop()->getlistenCnt();
        }
      }
    }
    m_metrics.m_pendingTasks.store(totalPending);
    m_metrics.m_cpuUsage.store(cpu);
    m_metrics.m_avgLatency.store(mem);
    // 触发调整检查
    adjustPoolSize();
    // 每 5 秒采集一次
    std::this_thread::sleep_for(std::chrono::seconds(5));
  }
}

double DynamicEventLoopThreadPool::getCpuUsage()
{
  static uint64_t lastTotal = 0;
  static uint64_t lastIdle = 0;
  
  std::ifstream stat("/proc/stat");
  if (!stat) {
    LOG_ERROR("get /proc/stat failed");
    return 0.0;
  }
  std::string line;
  std::getline(stat, line);
  std::istringstream iss(line);

  std::string cpu;

  uint64_t user, nice, system, idle;
  iss >> cpu >> user >> nice >> system >> idle;

  const uint64_t total = user + nice + system + idle;
  const uint64_t idleDiff = idle - lastIdle;
  const uint64_t totalDiff = total - lastTotal;
  lastIdle = idle;
  lastTotal = total;

  if (totalDiff > 0) {
    return 100.0 * (1.0 - (double)idleDiff / totalDiff);
  } else {
    return 0.0;
  }
}

int64_t DynamicEventLoopThreadPool::getMemUsage()
{
  std::ifstream statmFile("/proc/self/statm");
  if (!statmFile) {
    LOG_ERROR("get /proc/self/statm failed");
    return -1;
  }
  // 读取常驻内存大小
  uint64_t residentSetSize;
  statmFile >> residentSetSize; // 常驻内存大小（单位：页）
  // 获取系统页大小
  uint64_t pageSize = sysconf(_SC_PAGESIZE);
  // 将常驻内存大小转换为 KB
  return residentSetSize * pageSize / 1024;
}

void DynamicEventLoopThreadPool::adjustPoolSize()
{
  // 避免震荡
  static Timestamp lastAdjust;
  auto now = Timestamp::now();
  if (now - lastAdjust < 30) {
    return;
  }
  lastAdjust = now;
  // 获取当前所有的 loop 监听的连接数
  const int64_t pendingTasks = m_metrics.m_pendingTasks.load();
  // 获取当前线程数
  const int currentThread = m_currentThread.load();
  // 获取当前 cpu 使用率
  const double cpuUsage = m_metrics.m_cpuUsage.load();
  
  // 扩容条件：高负载且未达上限
  if (cpuUsage > 70.0 && pendingTasks > 100000 && currentThread < m_maxThread) {
    const int addNum = std::min(
      static_cast<int>(pendingTasks / 500), // 每 500个任务增加一个线程
      m_maxThread - currentThread
    );
    // 3. 扩容线程
    expandPool(addNum);
  }
  
  // 缩容条件：低负载且未达下限
  if (cpuUsage < 30.0 && pendingTasks <= 100000 && currentThread > m_minThread) {
    const int subNum = std::min(
      static_cast<int>(pendingTasks / 200), // 每 200个任务减少一个线程
      currentThread - m_minThread
    );
    // 4. 缩容线程
    shrinkPool(subNum);
  }
}


void DynamicEventLoopThreadPool::expandPool(const int addNum)
{
  std::lock_guard<std::mutex> lk(m_mutex);
  for (int i = 0; i < addNum; i++) {

    m_workers.emplace_back(
      new EventLoopThread(m_threadInitCallback, m_name),
      true);
    m_currentThread++;
    m_loops.emplace_back(m_workers.back().m_thread->startLoop());
  }
  LOG_INFO("Pool expanded to %d thread", m_currentThread.load());
}

void DynamicEventLoopThreadPool::shrinkPool(const int subNum)
{
  std::lock_guard<std::mutex> lk(m_mutex);
  int sub = 0;
  auto it = m_workers.begin();
  while (it != m_workers.end() && sub < subNum) {
    if ((*it).m_active.load()) {
      (*it).m_thread->stopLoop();
      (*it).m_active.store(false);
      it = m_workers.erase(it);
      sub++;
      m_workers.erase(it);
      --m_currentThread;
    } else {
      it++;
    }
  }
  LOG_INFO("Pool shrinked to %d threads", m_currentThread.load());
}


